package com.atguigu.dga.meta.service.impl;

import com.atguigu.dga.config.MetaConstant;
import com.atguigu.dga.meta.bean.TableMetaInfoExtra;
import com.atguigu.dga.meta.mapper.TableMetaInfoExtraMapper;
import com.atguigu.dga.meta.service.TableMetaInfoExtraService;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.apache.commons.lang3.RandomUtils;
import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Service;

import java.sql.Timestamp;
import java.util.List;
import java.util.stream.Collectors;

/**
 * <p>
 * 元数据表附加信息 服务实现类
 * </p>
 *
 * @author atguigu
 * @since 2023-10-27
 */
@Service
public class TableMetaInfoExtraServiceImpl extends ServiceImpl<TableMetaInfoExtraMapper, TableMetaInfoExtra> implements TableMetaInfoExtraService {

    @Autowired
    ApplicationContext context;
    @Override
    public void initMetaInfoExtra(String db, String assessDate) throws MetaException {
        HiveMetaStoreClient client = context.getBean(HiveMetaStoreClient.class);
        //1.为了保证幂等性，只为缺少了辅助信息的表生成信息
        List<String> allTables = client.getAllTables(db);

        //2.为缺少了辅助信息的表生成信息
        List<TableMetaInfoExtra> tableMetaInfoExtras = allTables
            .stream()
            //把缺少辅助信息的表过滤出来
            .filter(name -> {
                //查询当前表有没有辅助信息
                QueryWrapper<TableMetaInfoExtra> queryWrapper = new QueryWrapper<TableMetaInfoExtra>()
                    .eq("table_name", name)
                    .eq("schema_name", db);

                TableMetaInfoExtra metaInfoExtra = getOne(queryWrapper);
                return metaInfoExtra == null;
            })
            .map(name -> getTableMetaInfoExtraByName(name, db))
            .collect(Collectors.toList());

        //3.写入到数据库
        saveBatch(tableMetaInfoExtras);

        //4.关闭连接，节省资源
        client.close();
    }

    //根据表名生成TableMetaInfoExtra
    public TableMetaInfoExtra getTableMetaInfoExtraByName(String name,String db){
        TableMetaInfoExtra tableMetaInfoExtra = new TableMetaInfoExtra();
        tableMetaInfoExtra.setTableName(name);
        tableMetaInfoExtra.setSchemaName(db);
        //技术负责人和业务负责人，由工作人员自己录入。为了后续测试方便，模拟一些数据
        String [] tecOwners = {"张三","李四","王五","赵六","田七"};
        String [] busiOwners = {"张小三","李小四","王中五","赵大六","田巨七"};
        tableMetaInfoExtra.setTecOwnerUserName(tecOwners[RandomUtils.nextInt(0,tecOwners.length-1)]);
        tableMetaInfoExtra.setBusiOwnerUserName(busiOwners[RandomUtils.nextInt(0,busiOwners.length-1)]);
        //类似存储周期类型，这样的变量应该是作为配置的常量来引用
        //赋值初始值 UNSET
        tableMetaInfoExtra.setLifecycleType(MetaConstant.LIFECYCLE_TYPE_UNSET);
        tableMetaInfoExtra.setLifecycleDays(-1l);
        tableMetaInfoExtra.setSecurityLevel(MetaConstant.SECURITY_LEVEL_UNSET);
        tableMetaInfoExtra.setDwLevel(getDwLevelByName(name));
        tableMetaInfoExtra.setCreateTime(new Timestamp(System.currentTimeMillis()));
        return tableMetaInfoExtra;
    }

    /*
        根据名字，判断数仓的层级
            ods_
            dim_
            dwd_
            ads_
            dws_
     */
    public String getDwLevelByName(String name){

        //统一转换为大写
        String upperCase = name.toUpperCase();
        String prefix = upperCase.substring(0, 3);
        if (prefix.contains(MetaConstant.DW_LEVEL_ODS)){
            return  MetaConstant.DW_LEVEL_ODS;
        }else  if (prefix.contains(MetaConstant.DW_LEVEL_DWS)){
            return  MetaConstant.DW_LEVEL_DWS;
        }else  if (prefix.contains(MetaConstant.DW_LEVEL_DWD)){
            return  MetaConstant.DW_LEVEL_DWD;
        }else  if (prefix.contains(MetaConstant.DW_LEVEL_ADS)){
            return  MetaConstant.DW_LEVEL_ADS;
        }else  if (prefix.contains(MetaConstant.DW_LEVEL_DIM)){
            return  MetaConstant.DW_LEVEL_DIM;
        }else {
            return  MetaConstant.DW_LEVEL_OTHER;
        }
    }
}
